package com.bw.yk02.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

public class MyKafkaUtil {

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
        Properties properties = new Properties();

        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.18.105:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        FlinkKafkaConsumer<String> flinkKafkaConsumer=new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), properties);
        flinkKafkaConsumer.setStartFromEarliest();
        return flinkKafkaConsumer;
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.18.105:9092");
        return new FlinkKafkaProducer<>(topic, new SimpleStringSchema(), properties);
    }

    public static <T> FlinkKafkaProducer<JSONObject> getKafkaSinkBySchema() {

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.18.105:9092");

        KafkaSerializationSchema<JSONObject> kafkaSchema= new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                String topic="dwd_"+element.getString("table");
                String data= element.getString("data");
                //order_cargo
                return new ProducerRecord<>(topic,data.getBytes());
            }
        };

        //创建Kafka生产者对象并返回
        return new FlinkKafkaProducer<JSONObject>("default_topic",
                kafkaSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

}
